
package org.example.aaaatest.METAQ;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.taobao.metaq.client.MetaPushConsumer;

import java.util.List;


public class PushConsumer {


    public static void main(String[] args) throws InterruptedException, MQClientException {

        MetaPushConsumer consumer = new MetaPushConsumer("RebalanceTest_Consumer_Group");


        consumer.subscribe("TopicTest1", "TagA || TagC || TagD");
        consumer.setConsumeMessageBatchMaxSize(3);

        consumer.subscribe("TopicTest2", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {


            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                System.out.println(Thread.currentThread().getName() + " Rcs: " + msgs);
                // for (MessageExt msg : msgs) {
                // if (msg.getTags().equals("TagA")) {
                // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                // }
                // }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        /**
         * Consumer对象在使用之前必须要调用start初始化，初始化一次即可<br>
         */
        consumer.start();

        System.out.println("Consumer Started.");
    }
}